package com.yixiu.bi.kafka;

import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;


/**
 * Kafka生产者客户端工具
 *
 *建议：所有的配置文件路径，写到一个枚举类中。
 *
 * 注意：
 * (1)kafka发送消息会随机的发送到不同的分区，因此消息全局顺序是乱的。
 * 如果要求有序，创建topic时需要指定分区数为1，--partitions 1
 *
 * 参考：
 *(1)kafka java api    https://www.cnblogs.com/qizhelongdeyang/p/7354183.html
 *(2)kafka config   https://blog.csdn.net/jiecxy/article/details/53395336
 */
public class KafkaProducerUtil {
    private  static Logger logger  = LoggerFactory.getLogger(KafkaProducerUtil.class);

    // 使用这个静态变量保存每一个配置文件对应的kafaka生产者
    private static ConcurrentHashMap<String,KafkaProducerUtil> kafkaUtilMap  = new ConcurrentHashMap<String, KafkaProducerUtil>();

    private String topic = ""; // topic
    private KafkaProducer producer = null; // 生产者
    private  String configFile = ""; // 配置文件


    /**
     * 根据配置文件初始化KafkaProducerUtil对象
     * @param configFile 配置文件路径
     */
    private KafkaProducerUtil(String configFile) {
        this.configFile = configFile;
        // 读取配置，初始化producer 和 topic

        try {
            Properties kafkaProperties = new Properties();
            kafkaProperties.load(KafkaProducerUtil.class.getResourceAsStream(configFile));
            // 读取topic
            this.topic  = kafkaProperties.getProperty("topic");
            if(StringUtils.isEmpty(this.topic)) {
                logger.error( configFile + "没有配置topic");
                return;
            }
            // 用于建立与kafka集群的连接，这个list仅仅影响用于初始化的hosts，来发现全部的servers。
            //格式：host1:port1,host2:port2,…，数量尽量不止一个，以防其中一个down了
            String broker = kafkaProperties.getProperty("bootstrap.servers");
            if(StringUtils.isEmpty(broker)) {
                logger.error(configFile + "没有配置bootstrap.servers");
                return;
            }
            // 其他的配置信息，均采用Kafka客户端默认配置
            if(!kafkaProperties.containsKey("retries")) { // 默认重传3次
                // 限制客户端在单个连接上能够发送的未响应请求的个数。设置此值是1表示kafka broker在响应请求之前client不能再向同一个broker发送请求，但吞吐量会下降
                kafkaProperties.put("max.in.flight.requests.per.connection","1");
                kafkaProperties.put("retries","3");
            }
            if(!kafkaProperties.containsKey("key.serializer")) {
                kafkaProperties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            }
            if(!kafkaProperties.containsKey("value.serializer")) {
                kafkaProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            }

            // 初始化KafkaProducer，有可能异常
            producer = new KafkaProducer<String, String>(kafkaProperties);
        }catch (IOException ex) {
            logger.error("配置文件不存在:" + configFile,ex);
        } catch (Exception ex) {
            logger.error("创建KafkaProducerUtil异常:" + configFile,ex);
            this.producer = null;
        }
    }


    /**
     * 根据配置文件获取工具
     * kafkaProducerUtil.producer != null 说明获取成功，否则失败
     * @param configFile
     * @return kafkaProducerUtil.producer != null 说明获取成功，否则失败
     */
    private static KafkaProducerUtil getKafkaProducerUtil(String configFile) {
        if(kafkaUtilMap.containsKey(configFile)) {
            return kafkaUtilMap.get(configFile);
        }

        KafkaProducerUtil  kafkaProducerUtil  = new KafkaProducerUtil(configFile);
        if(kafkaProducerUtil.producer != null) {
            synchronized (kafkaUtilMap) {
                if(!kafkaUtilMap.containsKey(configFile)) {
                    kafkaUtilMap.put(configFile,kafkaProducerUtil);
                }
            }
        }
        return  kafkaProducerUtil;
    }


    /**
     * 根据配置文件，发送kafka消息
     * @param configFile 配置文件路径
     * @param message 消息
     * @return 发送失败返回null,否则返回发送结果Future<RecordMetadata>对象
     */
    public static   Future<RecordMetadata> send(String configFile,String message) {
        KafkaProducerUtil kafkaProducerUtil = getKafkaProducerUtil(configFile);
        if(kafkaProducerUtil.producer != null) {
            Future<RecordMetadata>  result = kafkaProducerUtil.producer.send(new ProducerRecord<String, String>(kafkaProducerUtil.topic, message));
            return  result;
        }
        return  null;
    }

    /**
     *发送缓冲区
     * @param configFile 配置文件路径
     * @return 刷新缓冲区是否成功，成功返回true
     */
    public static Boolean flush(String configFile) {
        KafkaProducerUtil kafkaProducerUtil = getKafkaProducerUtil(configFile);
        if(kafkaProducerUtil.producer != null) {
            kafkaProducerUtil.producer.flush();
            return  true;
        } else {
            return  false;
        }
    }

    /**
     * 关闭生产者
     * 注意：关闭之后就不能再发送了，再发送报错： Cannot send after the producer is closed.
     * 没有打开的方法，因此建议不调用关闭，等程序结束时自动关闭
     * @param configFile
     * @return
     */
    public static Boolean close(String configFile) {
        KafkaProducerUtil kafkaProducerUtil = getKafkaProducerUtil(configFile);
        if(kafkaProducerUtil.producer != null) {
            kafkaProducerUtil.producer.close();
            return  true;
        } else {
            return  false;
        }
    }


}
